package com.cike.sparkstudy.sql.scala

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object JSONDataSource {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local")
      .setAppName("JSONDataSource")

    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    //创建学生成绩的DataFrame
    val studentScoreDF = sqlContext.read.json("/developerCodes/test/jsonTest/students.json")
    //把DataFrame注册为临时表
    studentScoreDF.registerTempTable("students_score")
    //查询出成绩大于等于80分的学生
    val goodStudentScoresDF = sqlContext.sql("select name,score from students_score where score >= 80")
    //取出大于等于80分的学生的名字集合
    val goodStudentNames = goodStudentScoresDF.rdd.map{ row => row(0)}.collect()

    //创建学生基本信息DataFrame
    val studentInfoJsons = Array("{\"name\":\"Leo\", \"age\":18}",
      "{\"name\":\"Marry\", \"age\":17}",
      "{\"name\":\"Jack\", \"age\":19}")
    val studentInfoJsonsRDD = sc.parallelize(studentInfoJsons)
    val studentInfoJsonsDF = sqlContext.read.json(studentInfoJsonsRDD)

    //注册临时表并查询分数大于等于80分的学生信息
    studentInfoJsonsDF.registerTempTable("students_info")
    var sql = "select name,age from students_info where name in ("
    val goodStudentNamesLength = goodStudentNames.length
    for(i <- 0 until goodStudentNamesLength){
      sql += "'" + goodStudentNames(i) + "'"
      if(i < goodStudentNamesLength -1){
        sql += ","
      }
    }
    sql += ")"

    val goodStudentInfosDF = sqlContext.sql(sql)

    //将分数大于等于80的学生的成绩和基本信息进行join
    val goodStudentsRDD = goodStudentScoresDF
      .rdd.map{row => (row.getAs[String]("name"),row.getAs[Long]("score"))}
      .join(goodStudentInfosDF.rdd.map{row => (row.getAs[String]("name"),row.getAs[Long]("age"))})

    //将RDD转换成DataFrame
    val goodStudentsRDDRow = goodStudentsRDD.map(info => Row(info._1,info._2._1.toInt,info._2._2.toInt))

    //编程创建元数据
    val structType = StructType(Array(
      StructField("name", StringType, true),
      StructField("score", IntegerType, true),
      StructField("age", IntegerType, true)))

    val goodStudentsDF = sqlContext.createDataFrame(goodStudentsRDDRow,structType)

    //将DataFrame保存到文件夹
    goodStudentsDF.write.format("json").save("/developerCodes/test/jsonTest/good_students")

  }

}
